package com.sts.sink.impl;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.StrUtil;
import com.sts.jdbc.JdbcOperation;
import com.sts.jdbc.model.ColumnInfo;
import com.sts.sink.model.JdbcSinkConfig;
import com.sts.source.StsSource;
import com.sts.task.StsTask;

import java.sql.PreparedStatement;
import java.sql.Types;
import java.util.List;
import java.util.Map;

/**
 * @author lsl
 * @since 2025年07月04日 11:04
 */
public class PostgresqlSinkImpl extends JdbcSinkImpl {

    public PostgresqlSinkImpl(StsSource stsSource, JdbcSinkConfig jdbcSinkConfig, StsTask stsTask, JdbcOperation jdbcOperation) {
        super(stsSource, jdbcSinkConfig, stsTask, jdbcOperation);
    }

    @Override
    public void insertSetCellValue(PreparedStatement preparedStatement, List<ColumnInfo> columnInfoList, List<String> rowValueList) throws Exception {
        int size = columnInfoList.size();

        for (int i = 0; i < size; i++) {
            String typeName = columnInfoList.get(i).getTypeName();

            String columnValue = null;
            int rowSize = rowValueList.size();
            if (i < rowSize) {
                columnValue = rowValueList.get(i);
            }

            int index = i + 1;
            if (StrUtil.isNotEmpty(columnValue)) {
                columnValue = switch (typeName) {
                    case "DATE" -> DateUtil.formatDate(DateUtil.parse(columnValue));
                    case "TIME" -> DateUtil.formatTime(DateUtil.parse(columnValue));
                    case "TIMESTAMP" -> {
                        if (columnValue.length() > 26) {
                            columnValue = columnValue.substring(0, 26);
                        }
                        yield DateUtil.formatDateTime(DateUtil.parse(columnValue));
                    }
                    default -> columnValue;
                };
            } else {
                columnValue = null;
            }

            preparedStatement.setObject(index, columnValue, Types.OTHER);
        }
    }

    @Override
    public String buildUpsertSql(List<ColumnInfo> columnInfoList, Map<String, ColumnInfo> tableAllPrimaryKeyColumnMap) {
        String identifier = getJdbcSinkConfig().getJdbcConnectionInfo().getIdentifier();

        StringBuilder upsertSql = new StringBuilder("INSERT INTO ");
        upsertSql.append(identifier).append(getJdbcSinkConfig().getSchema()).append(identifier)
                .append(".")
                .append(identifier).append(getJdbcSinkConfig().getTable()).append(identifier)
                .append(" (");

        for (ColumnInfo columnInfo : columnInfoList) {
            upsertSql.append(identifier).append(columnInfo.getColumnName()).append(identifier).append(", ");
        }

        upsertSql.delete(upsertSql.lastIndexOf(","), upsertSql.length());
        upsertSql.append(") VALUES (");

        upsertSql.append("?, ".repeat(columnInfoList.size()));
        upsertSql.delete(upsertSql.lastIndexOf(","), upsertSql.length());

        upsertSql.append(") on conflict (");

        for (Map.Entry<String, ColumnInfo> entry : tableAllPrimaryKeyColumnMap.entrySet()) {
            ColumnInfo value = entry.getValue();
            upsertSql.append(identifier).append(value.getColumnName()).append(identifier).append(", ");
        }
        upsertSql.delete(upsertSql.lastIndexOf(","), upsertSql.length());

        upsertSql.append(") do update set ");

        for (ColumnInfo columnInfo : columnInfoList) {
            String columnName = columnInfo.getColumnName();

            // 不是主键的字段需要拼接在 on duplicate key update 后面
            if (!tableAllPrimaryKeyColumnMap.containsKey(columnName.toUpperCase())) {
                upsertSql.append(identifier).append(columnName).append(identifier).append(" = ?, ");
            }
        }

        upsertSql.delete(upsertSql.lastIndexOf(","), upsertSql.length());
        return upsertSql.toString();
    }

    @Override
    public void upsertSetCellValue(PreparedStatement preparedStatement, List<ColumnInfo> columnInfoList, List<String> rowValueList, Map<String, ColumnInfo> tableAllPrimaryKeyColumnMap) throws Exception {
        int size = columnInfoList.size();

        for (int i = 0; i < size; i++) {
            String typeName = columnInfoList.get(i).getTypeName();

            String columnValue = null;
            int rowSize = rowValueList.size();
            if (i < rowSize) {
                columnValue = rowValueList.get(i);
            }

            int index = i + 1;
            if (StrUtil.isNotEmpty(columnValue)) {
                columnValue = switch (typeName) {
                    case "DATE" -> DateUtil.formatDate(DateUtil.parse(columnValue));
                    case "TIME" -> DateUtil.formatTime(DateUtil.parse(columnValue));
                    case "TIMESTAMP" -> {
                        if (columnValue.length() > 26) {
                            columnValue = columnValue.substring(0, 26);
                        }
                        yield DateUtil.formatDateTime(DateUtil.parse(columnValue));
                    }
                    default -> columnValue;
                };
            } else {
                columnValue = null;
            }

            preparedStatement.setObject(index, columnValue, Types.OTHER);
        }

        int count = 0;
        for (int i = 0; i < size; i++) {
            ColumnInfo columnInfo = columnInfoList.get(i);
            if (tableAllPrimaryKeyColumnMap.containsKey(columnInfo.getColumnName().toUpperCase())) {
                continue;
            }

            String columnValue = null;
            int rowSize = rowValueList.size();
            if (i < rowSize) {
                columnValue = rowValueList.get(i);
            }

            if (StrUtil.isEmpty(columnValue)) {
                columnValue = null;
            }

            int index = size + count + 1;
            preparedStatement.setObject(index, columnValue, Types.OTHER);

            count++;
        }
    }
}
